package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.Voucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisConstants;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.SimpleRedisLock;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private ISeckillVoucherService secKillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private VoucherOrderServiceImpl voucherOrderService;
    @Autowired
    private RedissonClient redissonClient;

//    private static final BlockingQueue<VoucherOrder> arrayBlockingQueue = new ArrayBlockingQueue<>(1024*1024);
//    创建一个线程池
    private static  final ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
//    这个注解会在注入依赖之后 定义一些初始化的动作

    @PostConstruct
    private void init(){
        singleThreadExecutor.submit(new VoucherOrderHandler());
    }
//    private class VoucherOrderHandler implements Runnable{
////        线程的任务：从队列中取出一个订单对象 并且对数据库交互
//        @Override
//        public void run() {
//            while (true){
//                try {
//                    VoucherOrder voucherOrder = arrayBlockingQueue.take();
//                    handleVoucherOrder(voucherOrder);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//            }
//        }
//    }

    private class VoucherOrderHandler implements Runnable{
        String queueName="stream.orders";
        @Override
        public void run() {
            while (true){
                try {
//                    1. 获取消息队列中的订单信息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed()));
//                    判断是否获取成功
                    if (list==null||list.isEmpty()){
//                    如果获取失败说明没有消息,继续下一次循环
                        continue;
                    }
//                    如果获取成功,解析消息中的订单信息
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> value = entries.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
//                      解析完毕可以下单
                    handleVoucherOrder(voucherOrder);
//                    ACK确认
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", entries.getId());
                } catch (Exception e) {
                    log.error("处理订单异常",e);
                    handlePendingList();
                    e.printStackTrace();
                }
            }
        }

        private void handlePendingList() {
            while (true){
                try {
//                    1. 获取pendingList中的订单信息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create(queueName, ReadOffset.from("0")));
//                    判断是否获取成功
                    if (list==null||list.isEmpty()){
//                    如果获取失败说明pendlist中没有消息
                        break;
                    }
//                    如果获取成功,解析消息中的订单信息
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> value = entries.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
//                      解析完毕可以下单
                    handleVoucherOrder(voucherOrder);
//                    ACK确认
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", entries.getId());
                } catch (Exception e) {
                    log.error("处理pending-list异常",e);
                    e.printStackTrace();
                    continue;
                }
            }

        }
    }
    private void handleVoucherOrder(VoucherOrder voucherOrder) {

        //一人一单问题
//        获取到用户的ID
        Long id = voucherOrder.getUserId();
//        其实这里不需要锁机制 因为已经在Redis进行了处理 但是以防万一还是加锁
//      redis获取锁
        RLock lock = redissonClient.getLock("lock:order:" + id);

//        尝试获取锁 redisson中如果空参那么会不等待  获取失败直接返回false
        boolean isLock = lock.tryLock();

        if (!isLock) {
//            获取锁失败 表示之前已经下过单了
            log.error("您已下单");
            return  ;
        }
//        获取锁成功
        try {
             voucherOrderService.createVoucherOrder(voucherOrder);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }finally {
//            如果创建订单出现异常 释放锁
//            simpleRedisLock.unLock();
            lock.unlock();
        }

    }

    private static  final DefaultRedisScript<Long> SECKILL_SCRIPT;
    static{
        SECKILL_SCRIPT =new DefaultRedisScript<>();
//        设置脚本的位置
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }
//          我们可以通过阻塞队列来完成对数据库的添加订单操作


//    @Override
//    public Result secKillVoucher(Long voucherId) {
//        Long userId = UserHolder.getUser().getId();
//
////        1. 执行LUA脚本
//        Long result = stringRedisTemplate.execute(
//                SECKILL_SCRIPT,
//                Collections.emptyList(),
//                voucherId.toString(), userId.toString());
////        2. 判断结果是否为0
//        assert result != null;
//        int i = result.intValue();
//
////        3. 如果不是0直接返回异常信息
//        if (i!=0){
//            return Result.fail(i==1?"库存不足":"不能重复下单");
//        }
////        4. 如果是0 将优惠券id,用户id和订单id放入阻塞队列
////        TODO 将用户ID和订单ID放入阻塞队列
//
////       创建一个订单对象
//        VoucherOrder voucherOrder = new VoucherOrder();
//        voucherOrder.setVoucherId(voucherId);
//        long orderId = redisIdWorker.nextId("order");
//
//        voucherOrder.setId(orderId);
//        voucherOrder.setUserId(userId);
//
////        加入阻塞队列
//        arrayBlockingQueue.add(voucherOrder);
//        return Result.ok(0);
//    }


    @Override
    public Result secKillVoucher(Long voucherId) {
        //        获取数据库中的秒杀券信息
        SeckillVoucher seckillVoucher = secKillVoucherService.getById(voucherId);
        if (seckillVoucher==null) {
            return Result.fail("该秒杀活动不存在");
        }
//        首先获取秒杀券的开始时间和结束时间 还有库存信息
        LocalDateTime beginTime = seckillVoucher.getBeginTime();
        if (beginTime.isAfter(LocalDateTime.now())) {
//            秒杀活动尚未开始
            return Result.fail("秒杀活动尚未开始");
        }

        LocalDateTime endTime = seckillVoucher.getEndTime();
        if (endTime.isBefore(LocalDateTime.now())) {
//            秒杀活动已经结束

            return Result.fail("秒杀活动已经结束");
        }
//        获取用户ID
        Long userId = UserHolder.getUser().getId();
//        获取订单ID
        long orderId = redisIdWorker.nextId("order");
//        1. 执行LUA脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(),String.valueOf(orderId));
//        2. 判断结果是否为0
        assert result != null;
        int i = result.intValue();

//        3. 如果不是0直接返回异常信息
        if (i!=0){
            return Result.fail(i==1?"库存不足":"不能重复下单");
        }
//        加入阻塞队列

        return Result.ok(orderId);
    }


    //    @Override
//    @Transactional
//    public Result secKillVoucher(Long voucherId) {
////        获取数据库中的秒杀券信息
//        SeckillVoucher seckillVoucher = secKillVoucherService.getById(voucherId);
////        首先获取秒杀券的开始时间和结束时间 还有库存信息
//        LocalDateTime beginTime = seckillVoucher.getBeginTime();
//        if (beginTime.isAfter(LocalDateTime.now())) {
////            秒杀活动尚未开始
//            return Result.fail("秒杀活动尚未开始");
//        }
//
//        LocalDateTime endTime = seckillVoucher.getEndTime();
//        if (endTime.isBefore(LocalDateTime.now())) {
////            秒杀活动已经结束
//            return Result.fail("秒杀活动已经结束");
//        }
//        Integer stock = seckillVoucher.getStock();
//        if (stock==null||stock<1){
//            return Result.fail("秒杀券已经卖完了");
//        }
//        //一人一单问题
////        获取到用户的ID
//        Long id = UserHolder.getUser().getId();
//
////        注意 我们在加锁的时候这里用intern 而不是单单用toString 方法  因为toString 也是相当于新建了一个String 引用变量
////        多次新建 不会是同一个对象 而intern是从常量池中返回一个对象 多次新建是同一个对象
////        synchronized (id.toString().intern()) {
//////            注意这个锁必须加载函数外面 因为必须等事务提交后才能释放锁 否则还是会出现并发问题
////            return voucherOrderService.createVoucherOrder(voucherId);
////        }
////
//
//
////        但是以上方法不会解决集群下的多线程问题 所以我们用Redis来实现并发锁
////        SimpleRedisLock simpleRedisLock = new SimpleRedisLock(stringRedisTemplate,"voucherOrder");
//
////      redis获取锁
//        RLock lock = redissonClient.getLock("lock:order:" + id);
//
////       尝试获取锁 并且获取锁的连接时间
////        boolean isLock = simpleRedisLock.tryLock(1200);
////        尝试获取锁 redisson中如果空参那么会不等待  获取失败直接返回false
//        boolean isLock = lock.tryLock();
//
//        if (!isLock) {
////            获取锁失败 表示之前已经下过单了
//            return  Result.fail("您已下单");
//        }
////        获取锁成功
//        try {
//            return voucherOrderService.createVoucherOrder(voucherId);
//        } catch (Exception e) {
//            throw new RuntimeException(e);
//        }finally {
////            如果创建订单出现异常 释放锁
////            simpleRedisLock.unLock();
//            lock.unlock();
//        }
//
//    }


    @Transactional
    public  Result  createVoucherOrder(Long voucherId){

            Long id = UserHolder.getUser().getId();
            Integer count = query().eq("user_id", id).eq("voucher_id", voucherId).count();
            if (count > 0) {
//            说明用户已经买过了
                return Result.fail("用户已经购买过");
            }


            boolean isBuy = secKillVoucherService.update()
                    .setSql("stock=stock-1")
                    .eq("voucher_id", voucherId)
                    .gt("stock", 0)
                    .update();
            if (!isBuy) {
                return Result.fail("秒杀券已经卖完了");
            }
            VoucherOrder voucherOrder = new VoucherOrder();
//        通过工具类设置一个订单ID
            long orderId = redisIdWorker.nextId("order");
            voucherOrder.setId(orderId);
            voucherOrder.setUserId(id);
//        获取到秒杀券的ID
            voucherOrder.setVoucherId(voucherId);
//        将购买秒杀券的订单保存到数据库
            save(voucherOrder);
            return Result.ok(orderId);
    }
    @Transactional
    public  void  createVoucherOrder(VoucherOrder voucherOrder){


        Integer count = query().eq("user_id", voucherOrder.getUserId()).
                                eq("voucher_id", voucherOrder.getVoucherId()).count();
        if (count > 0) {
//            说明用户已经买过了

            return ;
        }

        boolean isBuy = secKillVoucherService.update()
                .setSql("stock=stock-1")
                .eq("voucher_id", voucherOrder.getVoucherId())
                .gt("stock", 0)
                .update();
        if (!isBuy) {
            return;
        }

//        通过工具类设置一个订单ID

//        获取到秒杀券的ID
//        将购买秒杀券的订单保存到数据库
        save(voucherOrder);
    }
}
